package mq.cinrad.kafka.app.producer;

import java.io.File;
import java.io.FilenameFilter;

import org.apache.commons.io.filefilter.AndFileFilter;
import org.apache.commons.io.filefilter.NotFileFilter;
import org.apache.commons.io.filefilter.PrefixFileFilter;
import org.apache.commons.io.filefilter.SuffixFileFilter;
import org.apache.commons.io.filefilter.WildcardFileFilter;

import mq.cinrad.kafka.app.config.KafkaWatchFileConfig;

public class KafkaWatchFileFilter implements FilenameFilter {

	private KafkaWatchFileConfig kafkaWatchFileConfig;

	private AndFileFilter andFileFilter;

	public KafkaWatchFileFilter(KafkaWatchFileConfig kafkaWatchFileConfig) {
		super();
		this.kafkaWatchFileConfig = kafkaWatchFileConfig;
		if (this.kafkaWatchFileConfig != null)
			buildAndFileFilter();
	}

	@Override
	public boolean accept(File dir, String name) {

		if (andFileFilter != null && andFileFilter.getFileFilters() != null
				&& andFileFilter.getFileFilters().size() > 0) {
			return andFileFilter.accept(dir, name);
		}

		return true;
	}

	public KafkaWatchFileConfig getKafkaWatchFileConfig() {
		return kafkaWatchFileConfig;
	}

	private void buildAndFileFilter() {

		andFileFilter = new AndFileFilter();
		if (kafkaWatchFileConfig.getPrefixFileFilter() != null
				&& kafkaWatchFileConfig.getPrefixFileFilter().size() > 0) {
			PrefixFileFilter pf = new PrefixFileFilter(kafkaWatchFileConfig.getPrefixFileFilter());
			andFileFilter.addFileFilter(pf);
		}

		if (kafkaWatchFileConfig.getNotPrefixFileFilter() != null
				&& kafkaWatchFileConfig.getNotPrefixFileFilter().size() > 0) {
			PrefixFileFilter npf = new PrefixFileFilter(kafkaWatchFileConfig.getNotPrefixFileFilter());
			andFileFilter.addFileFilter(new NotFileFilter(npf));
		}

		if (kafkaWatchFileConfig.getSuffixFileFilter() != null
				&& kafkaWatchFileConfig.getSuffixFileFilter().size() > 0) {
			SuffixFileFilter sf = new SuffixFileFilter(kafkaWatchFileConfig.getSuffixFileFilter());
			andFileFilter.addFileFilter(sf);
		}

		if (kafkaWatchFileConfig.getNotSuffixFileFilter() != null
				&& kafkaWatchFileConfig.getNotSuffixFileFilter().size() > 0) {
			SuffixFileFilter nsf = new SuffixFileFilter(kafkaWatchFileConfig.getNotSuffixFileFilter());
			andFileFilter.addFileFilter(new NotFileFilter(nsf));
		}

		if (kafkaWatchFileConfig.getWildcardFileFilter() != null
				&& kafkaWatchFileConfig.getWildcardFileFilter().size() > 0) {
			WildcardFileFilter wf = new WildcardFileFilter(kafkaWatchFileConfig.getWildcardFileFilter());
			andFileFilter.addFileFilter(wf);
		}

		if (kafkaWatchFileConfig.getNotWildcardFileFilter() != null
				&& kafkaWatchFileConfig.getNotWildcardFileFilter().size() > 0) {
			WildcardFileFilter nwf = new WildcardFileFilter(kafkaWatchFileConfig.getNotWildcardFileFilter());
			andFileFilter.addFileFilter(new NotFileFilter(nwf));
		}

	}

}
